06 错误处理与日志

你的Agent API上线了，用户开始调用了。然后某一天，Agent调用大模型超时了，返回了一个HTML格式的500错误页面——前端拿到这个HTML直接解析失败，整个页面白屏。

这就是没有做好错误处理的后果。对于API服务来说，任何错误都应该返回统一格式的JSON，而不是Flask默认的HTML错误页面。

同时，你需要记录日志——哪些请求出错了、错误原因是什么、哪个用户遇到了问题。没有日志，出了问题就像蒙着眼睛找Bug。

一、默认错误行为

Flask内置了一套HTTP异常：400（请求错误）、401（未授权）、403（禁止访问）、404（未找到）、405（方法不允许）、500（服务器错误）等。

默认情况下，这些异常会返回HTML格式的错误页面。对于API服务，我们需要把它们改成JSON格式。

二、注册错误处理器

用@app.errorhandler()装饰器注册自定义的错误处理函数：

python

from flask import Flask, jsonify

app = Flask(__name__)


@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": str(error.description)}), 400


@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "接口不存在"}), 404


@app.errorhandler(405)
def method_not_allowed(error):
    return jsonify({"error": "请求方法不允许"}), 405


@app.errorhandler(500)
def internal_error(error):
    return jsonify({"error": "服务器内部错误"}), 500

现在所有错误都会返回JSON格式：

json

{"error": "接口不存在"}

注意：错误处理函数的返回值必须手动设置状态码（第二个值），否则会默认返回200。

2.1 按异常类注册

除了用状态码，也可以用异常类来注册：

python

from werkzeug.exceptions import BadRequest, NotFound

@app.errorhandler(BadRequest)
def handle_bad_request(error):
    return jsonify({"error": str(error.description)}), 400

@app.errorhandler(NotFound)
def handle_not_found(error):
    return jsonify({"error": "接口不存在"}), 404

状态码和异常类可以互换使用，因为BadRequest.code == 400。

2.2 捕获所有HTTP异常

如果想用一个处理器处理所有HTTP错误，可以注册HTTPException：

python

from flask import json
from werkzeug.exceptions import HTTPException

@app.errorhandler(HTTPException)
def handle_http_exception(error):
    """所有HTTP错误统一返回JSON"""
    response = error.get_response()
    response.data = json.dumps({
        "code": error.code,
        "name": error.name,
        "description": error.description,
    })
    response.content_type = "application/json"
    return response

这种方式最省事——一个函数搞定所有HTTP错误。

2.3 捕获未知异常

除了HTTP异常，代码中还可能出现其他异常（比如数据库连接失败、第三方API超时）。可以注册Exception处理器：

python

@app.errorhandler(Exception)
def handle_exception(error):
    # HTTP异常走已注册的处理器
    if isinstance(error, HTTPException):
        return error

    # 其他异常：记录日志，返回500
    app.logger.error(f"未处理的异常: {error}", exc_info=True)
    return jsonify({"error": "服务器内部错误"}), 500

关键点：先判断是不是HTTPException，是的话直接return让它走已注册的处理器。只有非HTTP异常才走这里的逻辑。

三、自定义异常类

对于Agent API，你可能需要更丰富的错误信息——不只是一个message，还要有错误码、详细信息等。可以定义自定义异常类：

python

from flask import jsonify


class AgentError(Exception):
    """Agent API的基础异常"""
    status_code = 400

    def __init__(self, message, status_code=None, payload=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code
        self.payload = payload

    def to_dict(self):
        rv = dict(self.payload or ())
        rv["error"] = self.message
        rv["code"] = self.status_code
        return rv

注册异常处理器：

python

@app.errorhandler(AgentError)
def handle_agent_error(error):
    return jsonify(error.to_dict()), error.status_code

在代码中使用：

python

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    if not data:
        raise AgentError("请求体不能为空")

    message = data.get("message")
    if not message:
        raise AgentError("缺少message字段")

    session_id = data.get("session_id")
    if session_id and len(session_id) > 100:
        raise AgentError(
            "session_id过长",
            status_code=400,
            payload={"max_length": 100},
        )

    return {"reply": f"收到: {message}"}

返回的错误格式：

json

{
    "error": "session_id过长",
    "code": 400,
    "max_length": 100
}

四、abort函数

有时候你不需要自定义异常类，只是想快速返回一个错误。abort函数可以直接抛出HTTP异常：

python

from flask import abort

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    if not data or "message" not in data:
        abort(400, description="缺少message字段")

    return {"reply": "收到"}

abort(400)会立即停止执行，抛出一个BadRequest异常，然后被你注册的错误处理器捕获。

description参数会成为error.description，在错误处理器中可以通过str(error.description)获取。

五、Blueprint错误处理

Blueprint也可以有自己的错误处理器：

python

chat_bp = Blueprint("chat", __name__)

@chat_bp.errorhandler(429)
def rate_limit_exceeded(error):
    return jsonify({"error": "请求太频繁，请稍后再试"}), 429

Blueprint的错误处理器只在该蓝图的视图函数中触发。如果蓝图内部没有匹配的处理器，会向上查找应用级别的处理器。

一个实用的做法：API接口按前缀区分错误格式：

python

@app.errorhandler(404)
@app.errorhandler(405)
def handle_api_error(error):
    if request.path.startswith("/api/"):
        return jsonify({"error": str(error.description)}), error.code
    else:
        return error  # 非API路径返回默认HTML

六、日志基础

Flask内置了Python标准的logging模块，通过app.logger访问：

python

app.logger.debug("调试信息")
app.logger.info("一般信息")
app.logger.warning("警告信息")
app.logger.error("错误信息")
app.logger.critical("严重错误")

五个级别从低到高：DEBUG < INFO < WARNING < ERROR < CRITICAL。低于配置级别的日志会被忽略。

6.1 配置日志

默认情况下，Flask的日志级别是WARNING，只有警告和错误才会输出。开发阶段建议改成DEBUG：

python

import logging

# 设置日志级别
app.logger.setLevel(logging.DEBUG)

更推荐用dictConfig做完整配置：

python

from logging.config import dictConfig

dictConfig({
    "version": 1,
    "formatters": {
        "default": {
            "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "stream": "ext://sys.stderr",
            "formatter": "default",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["console"],
    },
})

app = Flask(__name__)

日志输出格式：

[2026-01-15 10:30:45] INFO in chat: 用户 user_123 发送了消息
[2026-01-15 10:30:46] ERROR in chat: Agent调用超时

6.2 在请求中记录日志

Agent API中，记录每次请求的关键信息很有用：

python

@app.route("/chat", methods=["POST"])
def chat():
    data = request.get_json()
    message = data.get("message", "")
    session_id = data.get("session_id", "default")

    app.logger.info(f"收到请求: session={session_id}, message={message[:50]}")

    try:
        # Agent处理逻辑...
        reply = f"收到: {message}"
    except Exception as e:
        app.logger.error(f"Agent处理失败: {e}", exc_info=True)
        abort(500)

    app.logger.info(f"回复完成: session={session_id}")
    return {"reply": reply}

exc_info=True会把完整的异常堆栈记录到日志中，方便排查问题。

6.3 注入请求信息到日志

自定义Formatter，在日志中自动加入请求的URL和IP：

python

from flask import has_request_context, request


class RequestFormatter(logging.Formatter):
    def format(self, record):
        if has_request_context():
            record.url = request.url
            record.remote_addr = request.remote_addr
            record.method = request.method
        else:
            record.url = None
            record.remote_addr = None
            record.method = None

        return super().format(record)


formatter = RequestFormatter(
    "[%(asctime)s] %(remote_addr)s %(method)s %(url)s\n"
    "%(levelname)s in %(module)s: %(message)s"
)

日志输出变成：

[2026-01-15 10:30:45] 127.0.0.1 POST http://localhost:5000/api/chat
INFO in chat: 收到请求

每条日志都带着请求的IP、方法和URL，排查问题时一目了然。

七、错误处理 + 日志组合

把错误处理和日志结合起来，形成完整的错误管理体系：

python

from flask import Flask, jsonify
from werkzeug.exceptions import HTTPException
import logging

app = Flask(__name__)


class AgentError(Exception):
    status_code = 400

    def __init__(self, message, status_code=None):
        super().__init__()
        self.message = message
        if status_code is not None:
            self.status_code = status_code


# 1. 自定义业务异常：记录warning日志，返回JSON
@app.errorhandler(AgentError)
def handle_agent_error(error):
    app.logger.warning(f"业务错误: {error.message}")
    return jsonify({"error": error.message, "code": error.status_code}), error.status_code


# 2. HTTP异常：返回JSON
@app.errorhandler(HTTPException)
def handle_http_exception(error):
    return jsonify({
        "error": error.description,
        "code": error.code,
    }), error.code


# 3. 未知异常：记录error日志（含堆栈），返回500
@app.errorhandler(Exception)
def handle_exception(error):
    if isinstance(error, HTTPException):
        return error

    app.logger.error(f"未处理异常: {error}", exc_info=True)
    return jsonify({"error": "服务器内部错误", "code": 500}), 500

三层处理逻辑：

异常类型	日志级别	响应格式
`AgentError`（业务错误）	WARNING	JSON + 自定义消息
`HTTPException`（HTTP错误）	无	JSON + 标准描述
`Exception`（未知错误）	ERROR + 堆栈	JSON + "服务器内部错误"

八、生产环境日志建议

8.1 日志输出到文件

开发时输出到终端就够了，生产环境需要写入文件：

python

from logging.config import dictConfig

dictConfig({
    "version": 1,
    "formatters": {
        "default": {
            "format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
        },
    },
    "handlers": {
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "logs/app.log",
            "maxBytes": 10 * 1024 * 1024,  # 10MB
            "backupCount": 5,
            "formatter": "default",
        },
    },
    "root": {
        "level": "INFO",
        "handlers": ["file"],
    },
})

RotatingFileHandler会自动轮转日志文件——超过10MB就新建文件，最多保留5个备份。

8.2 第三方库日志

Agent项目中，LangChain、OpenAI等库也会产生日志。如果你想看它们的日志：

python

# 给特定库设置日志级别
logging.getLogger("langchain").setLevel(logging.INFO)
logging.getLogger("openai").setLevel(logging.WARNING)

或者把所有库的日志都收集起来：

python

root = logging.getLogger()
root.setLevel(logging.INFO)

8.3 Sentry集成

生产环境中，光有日志还不够。推荐用Sentry做错误监控——它能自动捕获未处理的异常，聚合重复错误，发送告警邮件：

bash

pip install sentry-sdk[flask]

python

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[FlaskIntegration()],
)

加了这两行代码后，所有导致500的异常都会自动上报到Sentry。你可以在Sentry的控制台上看到完整的堆栈、请求信息、发生频率。

九、总结

错误处理和日志是Agent API上线的必备条件：

统一错误格式：@app.errorhandler()把所有错误转成JSON
自定义异常：AgentError携带业务错误码和详细信息
abort快速中断：参数校验失败时直接abort(400)
日志分级：DEBUG/INFO/WARNING/ERROR，按需输出
请求信息注入：每条日志带上IP、URL、方法
生产环境：日志写文件 + Sentry错误监控

有了错误处理和日志，你的Agent API就不再是"出了问题两眼一抹黑"的状态了。

在下一篇文章中，我们将学习Gunicorn + Nginx部署——把你的Agent API从开发环境搬到生产服务器上。

06 错误处理与日志 ​

一、默认错误行为 ​

二、注册错误处理器 ​

2.1 按异常类注册 ​

2.2 捕获所有HTTP异常 ​

2.3 捕获未知异常 ​

三、自定义异常类 ​

四、abort函数 ​

五、Blueprint错误处理 ​

六、日志基础 ​

6.1 配置日志 ​

6.2 在请求中记录日志 ​

6.3 注入请求信息到日志 ​

七、错误处理 + 日志组合 ​

八、生产环境日志建议 ​

8.1 日志输出到文件 ​

8.2 第三方库日志 ​

8.3 Sentry集成 ​

九、总结 ​

06 错误处理与日志

一、默认错误行为

二、注册错误处理器

2.1 按异常类注册

2.2 捕获所有HTTP异常

2.3 捕获未知异常

三、自定义异常类

四、abort函数

五、Blueprint错误处理

六、日志基础

6.1 配置日志

6.2 在请求中记录日志

6.3 注入请求信息到日志

七、错误处理 + 日志组合

八、生产环境日志建议

8.1 日志输出到文件

8.2 第三方库日志

8.3 Sentry集成

九、总结